package com.fudan.run.trans;

import java.util.List;
import java.util.Map;

import org.apache.spark.api.java.JavaRDD;

import com.fudan.cfg.base.MapUtils;
import com.fudan.cfg.trans.NewFieldTransDef;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.TransAdapter;

@TransAdapter("new_field")
public class NewFieldTrans  extends TransHandler<NewFieldTransDef>{

	@Override
	public void trans(JobRunner jobRunner, NewFieldTransDef transNode, VariableRddMap variableMap) {
		String target = transNode.getTarget();
		JavaRDD<Map<String, Object>> targetRdd = variableMap.getRdd(target);
		JavaRDD<Map<String, Object>> afterRdd = targetRdd.map(l->{
			List<Map<String,Object>> fieldList = transNode.getFieldList();
			if(fieldList != null) {
				for (Map<String, Object> fieldDef : fieldList) {
					String type = MapUtils.getMapValue(fieldDef, "type");
					switch(type) {
						case "concat": {
							List<String> concatFields = MapUtils.getMapValue(fieldDef, "fields");
							String fieldName = MapUtils.getMapValue(fieldDef, "field_name");
							StringBuilder v = new StringBuilder();
							for (String f : concatFields) {
								v.append(MapUtils.getMapValue(l, f, ""));
							}
							l.put(fieldName, v.toString());
							break;
						}
						case "new": {
							String fieldName = MapUtils.getMapValue(fieldDef, "field_name");
							String value = MapUtils.getMapValue(fieldDef, "value");
							l.put(fieldName, value);
							break;
						}
						default:{
							break;
						}
					}
				}
			}
			return l;
		});
		variableMap.addRdd(transNode.getName(), afterRdd);
	}

}
